package com.atguigu.Func;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.Util.DruidDSUtil;
import com.atguigu.Util.JdbcUtil;
import com.atguigu.bean.TableProcess;
import com.atguigu.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * @author hjy
 * @create 2023/3/8 18:06
 */
public class DimTableProcessFunctiom extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private HashMap<String, TableProcess> hashMap;

    @Override
    public void open(Configuration parameters) throws Exception {
        hashMap = new HashMap<String, TableProcess>();
        //拿到的数据放入hashMap，防止状态加载有延迟
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall_config?" +
                "user=root&password=123456&useUnicode=true&" +
                "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        //从mysql拿数据 自定义一个util类
        List<TableProcess> jdbcUtilList = JdbcUtil.getJdbcUtil(conn, "select * from table_process where sink_type='dim'", TableProcess.class, true);
        for (TableProcess tableProcess : jdbcUtilList) {
            //todo 在这里也要进行一次建表操作，否则数据没被过滤掉但是没有地方可去
            createTable(tableProcess);
            String table = tableProcess.getSourceTable();
            //这里将从mysql的数据存放入hashMap中,方便后续的使用-->和状态的使用方式类似，同样是通过key得到value的值-->在processElement这个方法中
            hashMap.put(table, tableProcess);
        }
    }

    public DimTableProcessFunctiom(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    //value
    // {"before":null,"after":{"source_table":"base_category3","sink_table":"dim_base_category3","sink_columns":"id,name,category2_id","sink_pk":"id","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1669162876406,"snapshot":"false","db":"gmall-220623-config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1669162876406,"transaction":null}
    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {

        //获取状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        JSONObject jsonObject = JSONObject.parseObject(value);

        if ("d".equals(jsonObject.getString("op"))) {
            //1.转换为JavaBean
            TableProcess tableProcess = JSONObject.parseObject(JSONObject.parseObject(value).getString("before"), TableProcess.class);
            if ("dim".equals(tableProcess.getSinkType())) {
                broadcastState.remove(tableProcess.getSourceTable());
                hashMap.remove(tableProcess.getSourceTable());
            }
        } else {
            TableProcess tableProcess = JSONObject.parseObject(jsonObject.getString("after"), TableProcess.class);

            //2.建表
            createTable(tableProcess);

            //3.将数据写入状态
            if ("dim".equals(tableProcess.getSinkType())) {
                String key = tableProcess.getSourceTable();
                broadcastState.put(key, tableProcess);
            }
        }
    }

    //建表:create table if not exists db.tn(id varchar primary key ,name varchar ,sex varchar) xxx
    private void createTable(TableProcess tableProcess) {
        String sinkPk = tableProcess.getSinkPk();
        if (sinkPk == null || "".equals(sinkPk)) {
            sinkPk = "id";
        }
        String sinkExtend = tableProcess.getSinkExtend();
        if (sinkExtend == null) {
            sinkExtend = "";
        }
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append("create table if not exists ")
                .append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(tableProcess.getSinkTable())
                .append("(");

        String[] Columns = tableProcess.getSinkColumns().split(",");

        for (int i = 0; i < Columns.length; i++) {
            String column = Columns[i];
            stringBuilder.append(column).append(" varchar ");
            if (sinkPk.equals(column)) {
                stringBuilder.append(" primary key ");
            }
            if (i < Columns.length - 1) {
                stringBuilder.append(",");
            } else {
                stringBuilder.append(")")
                        .append(sinkExtend);
            }
        }
        String sql = stringBuilder.toString();
        System.out.println("建表语句" + sql);

        //写入phoenix
        DruidPooledConnection phoenixConn = null;
        try {
            phoenixConn = DruidDSUtil.getPhoenixConn();
            //预编译
            PreparedStatement preparedStatement = phoenixConn.prepareStatement(sql);
            //执行
            preparedStatement.execute();
        } catch (SQLException e) {
            throw new RuntimeException("建表" + tableProcess.getSinkTable() + "失败！！！");
        }
    }


    //{"database":"gmall-220623-flink","table":"comment_info","type":"insert","ts":1669162958,"xid":1111,"xoffset":13941,"data":{"id":1595211185799847960,"user_id":119,"nick_name":null,"head_img":null,"sku_id":31,"spu_id":10,"order_id":987,"appraise":"1204","comment_txt":"评论内容：48384811984748167197482849234338563286217912223261","create_time":"2022-08-02 08:22:38","operate_time":null}}
    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // 读取数据
        //过滤数据
        //行过滤+列过滤
        //写出

        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        String table = value.getString("table");
        String type = value.getString("type");
        JSONObject fields = value.getJSONObject("data");
        Set<Map.Entry<String, Object>> entries = fields.entrySet();
        TableProcess tableProcess = broadcastState.get(table);
        TableProcess mapTableProcess = hashMap.get(table);

        System.out.println(mapTableProcess);

        //行过滤
        if ((tableProcess != null || mapTableProcess != null) && ("bootstrap-insert".equals(type) || "insert".equals(type) || "update".equals(type))) {
            if (tableProcess == null) {
                tableProcess = mapTableProcess;
            }
            //列过滤
            String[] columns = tableProcess.getSinkColumns().split(",");
            List<String> columnsList = Arrays.asList(columns);
            Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
            while (iterator.hasNext()) {
                Map.Entry<String, Object> entry = iterator.next();
                if (!columnsList.contains(entry.getKey())) {
                    iterator.remove();
                }
            }
            value.put("sink_table", tableProcess.getSinkTable());
            out.collect(value);
        } else {
            System.out.println("这个key" + table + "不存在！ 或者操作类型为" + type);
        }

    }
}
